package com.atguigu.tingshu.album.receiver;

import com.alibaba.fastjson.JSON;
import com.atguigu.tingshu.album.service.AlbumStatService;
import com.atguigu.tingshu.common.kafka.constant.KafkaConst;
import com.atguigu.tingshu.vo.album.AlbumStatMqVo;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

/**
 * 专辑统计消息消费者
 */
@Slf4j
@Component
public class AlbumStatReceiver {

    @Autowired
    private AlbumStatService albumStatService;

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    /**
     * 消费专辑统计更新消息
     *
     * @param record Kafka 消息记录
     */
    @KafkaListener(topics = KafkaConst.TOPIC_ALBUM_STAT_UPDATE, groupId = KafkaConst.GROUP_ALBUM)
    public void consumeAlbumStatUpdate(ConsumerRecord<String, String> record, Acknowledgment acknowledgment) {
        try {
            String value = record.value();
            log.info("收到专辑统计更新消息 - Topic: {}, Partition: {}, Offset: {}, Message: {}",
                    record.topic(), record.partition(), record.offset(), value);

            // 解析消息
            AlbumStatMqVo albumStatMqVo = JSON.parseObject(value, AlbumStatMqVo.class);

            // 幂等性校验：使用 businessNo 防止重复消费
            String idempotentKey = "album:stat:update:" + albumStatMqVo.getBusinessNo();
            Boolean success = redisTemplate.opsForValue().setIfAbsent(idempotentKey, "1", 10, TimeUnit.MINUTES);

            if (Boolean.FALSE.equals(success)) {
                log.warn("专辑统计更新消息重复消费，已跳过 - BusinessNo: {}", albumStatMqVo.getBusinessNo());
                // 重复消费也需要手动提交 offset
                acknowledgment.acknowledge();
                return;
            }

            // 更新专辑统计信息
            albumStatService.updateAlbumStat(
                    albumStatMqVo.getAlbumId(),
                    albumStatMqVo.getStatType(),
                    albumStatMqVo.getCount());

            // 手动提交 offset
            acknowledgment.acknowledge();

            log.info("专辑统计更新成功 - AlbumId: {}, StatType: {}, Count: {}",
                    albumStatMqVo.getAlbumId(), albumStatMqVo.getStatType(), albumStatMqVo.getCount());

        } catch (Exception e) {
            log.error("消费专辑统计更新消息失败 - Error: {}", e.getMessage(), e);
            // 不提交 offset，Kafka 会在下次拉取时重新消费该消息
            // 如果需要立即重试，可以抛出异常
            // throw new RuntimeException("消费专辑统计更新消息失败", e);
        }
    }
}
